package stream

import (
	"context"
	"errors"
	"github.com/kataras/iris/v12"
	"github.com/spf13/pflag"
	"github.com/spf13/viper"
	"github.com/wutianze/nats.go"
	"gluenet/modules/manager/stream/aggregation"
	"gluenet/modules/manager/stream/handler/stream"
	"gluenet/modules/manager/stream/handler/syncer"
	"gluenet/pkg/apis"
	"gluenet/pkg/log/v1"
	"gluenet/pkg/nets"
	"gluenet/pkg/ops"
	"sync"
	"time"
)

type StorageServer struct {
	ctx    context.Context
	logger v1.Logger

	nets nets.GlueNets
	db   *ops.DefaultDB

	//prom string
	port string

	fs map[string]*nats.Subscription

	metricChan chan []byte

	metricSubscriber      *sync.Map
	metricSubscriberTotal *sync.Map

	StatusSubscriber      *sync.Map
	StatusSubscriberTotal *sync.Map

	infoflush *apis.Gflush

	metricCache *sync.Map
	stateCache  *sync.Map
}

func (r *StorageServer) InitFlag(flags *pflag.FlagSet) {
	flags.String("server.port", "10001", "gluenet server port")
	r.db.InitFlag(flags)
	r.nets.InitFlag(flags)
}

func (r *StorageServer) ViperConfig(viper *viper.Viper) {
}

func (r *StorageServer) InitViper(viper *viper.Viper) {
	r.port = viper.GetString("server.port")
	r.db.InitViper(viper)
	r.nets.InitViper(viper)
}

func (r *StorageServer) Initialize() {
	r.logger = v1.NewLogger("runner")

	if err := r.db.Connect(); err != nil {
		r.logger.Fatalf("data base connect %v", err)
	}
	if err := r.nets.Connect(); err != nil {
		r.logger.Fatalf("nets connect %v", err)
	}

	r.infoflush.Run(r.logger)

	app := iris.New()

	sctx := apis.StorageContext{
		Context:                r.ctx,
		Logger:                 r.logger,
		Rpc:                    r.nets,
		DataBase:               r.db,
		MetricsChan:            r.metricChan,
		MetricsSubscriber:      r.metricSubscriber,
		MetricsSubscriberTotal: r.metricSubscriberTotal,
		StatusSubscriber:       r.StatusSubscriber,
		StatusSubscriberTotal:  r.StatusSubscriberTotal,
		InfoFlusher:            r.infoflush,
		MetricCache:            r.metricCache,
		StateCache:             r.stateCache,
	}

	if fs, err := Register(r.nets, app, &sctx); err != nil {
		r.logger.Fatalf("register func  %v", err)
	} else {
		r.fs = fs
	}

	initStream(&sctx)
	go stream.StreamAllocator(&sctx)
	go syncer.SyncInfo(&sctx)
	go aggregation.Group(&sctx)

	if err := app.Run(iris.Addr(":" + r.port)); err != nil {
		if errors.Is(err, iris.ErrServerClosed) {
			r.logger.Infof("server closed")
		} else {
			r.logger.Infof("server error %v", err)
		}
	}
	//r.metricChan <- []byte("ppp")

	r.logger.Infof("server initialize finished")

	<-r.ctx.Done()
}
func (r *StorageServer) Close() {
	for k, v := range r.fs {
		if err := v.Unsubscribe(); err != nil {
			r.logger.Errorf("close nets push %v %v", k, err)
		}
	}

	if err := r.infoflush.Close(); err != nil {
		r.logger.Errorf("close glue.impl %v", err)
	}

	if err := r.db.Close(); err != nil {
		r.logger.Errorf("close glue.impl %v", err)
	}
	if err := r.nets.Close(); err != nil {
		r.logger.Errorf("close glue.rpc %v", err)
	}
}

func NewServer(ctx context.Context) *StorageServer {
	r := StorageServer{
		ctx:  ctx,
		db:   ops.NewDB(ctx, "glue.impl"),
		nets: nets.NewNets(ctx, "glue.rpc"),

		metricChan:            make(chan []byte),
		metricSubscriber:      &sync.Map{},
		metricSubscriberTotal: &sync.Map{},

		StatusSubscriber:      &sync.Map{},
		StatusSubscriberTotal: &sync.Map{},

		metricCache: &sync.Map{},
		stateCache:  &sync.Map{},

		infoflush: apis.NewGflush(ctx, time.Second*10),

		fs: make(map[string]*nats.Subscription),
	}
	return &r
}
